// -*-mode:c++; coding:utf-8-*-

#include <iostream>
#include <sstream>
#include <string>

using namespace std;

#include <boost/asio.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/thread/mutex.hpp>

#include <xbase_utility.hpp>
#include <naming_strategy.hpp>

#include "io_service_pool_impl.hpp"
#include "session_manager.hpp"
#include "scanner.hpp"
#include "scan_struct.hpp"

using namespace xbase;

typedef boost::system::error_code error_code;

unsigned scan_row_count = 0;

void on_get_next_row(scanner::ptr_type s_ptr, row_result::ptr_type row_ptr, const error_code &ec)
{
	if(ec)
	{
		if (ec == xbase::error::eos)
		{
			std::cout << "scan finished, row count: " << scan_row_count <<endl;
		}
		else
		{
			std::cout << "get next row failed : " << ec
				<< " - " << ec.message() << std::endl;
		}
	}
	else
	{
/*
		cout << "row: {row_name: " << dump_string_hex(row_ptr->m_row_name) << " columns: {";

		row_result::cell_vector::iterator iter = row_ptr->m_cell_vector.begin();
		while (iter != row_ptr->m_cell_vector.end())
		{
			cell &c = *iter;
			unsigned version_count = c.get_version_count();

			std::cout << "{column_name: " << c.m_column_name;
			std::cout << " version_count: " << version_count << " versions: {";

			unsigned show_version_num = min((unsigned)3, version_count);
			for (unsigned i=0; i<show_version_num; i++)
			{
				cell::version v;
				c.get_version(i, v);
				cout << "{" << v.first << "," << dump_string_hex(v.second) << "}, ";
			}
			cout << "}} ";

			++iter;
		}
		cout << "}}" << endl;
*/

		++scan_row_count;

		if (scan_row_count%500 == 0)
		{
			cout << "scan_row_count: " << scan_row_count << endl;
		}

		s_ptr->async_get_next_row(boost::bind(on_get_next_row, s_ptr, _1, _2));
	}
}

int main(int argc, char **argv)
{
	if(argc < 7)
	{
		cout <<argv[0]<<" master_ip master_port table_name start_row end_row column_names"<< endl;
		return 0;
	}

	LOG_CONFIG("../../conf/log4cplus.test.conf");

	boost::asio::ip::tcp::endpoint endpoint(
		boost::asio::ip::address::from_string(argv[1]),
		::atoi(argv[2])
		);

	string table_name(argv[3]);
	string start_row(argv[4]);
	string end_row(argv[5]);
	vector<string> column_names;
	
	for (int i=6;i<argc;i++)
	{
		column_names.push_back(argv[i]);
	}

	//
	// 首先初始化service_pool并启动
	// 

	scan_info si;
	si.m_table_name = table_name;
	si.m_start_row = start_row;
	si.m_end_row = end_row;
	si.m_column_names = column_names;

	boost::shared_ptr<io_service_pool> pool(new impl::single_io_service());
	client_options options;
	pool->set_options(options);
	pool->run();

	session_manager sm(*pool);
	root_meta_cache rmc(*pool, endpoint, sm);
	location_cache lc(*pool, rmc, sm);

	cout<<" table_name: "<<table_name<<" start_row: "<<start_row<<" end_row: "<<end_row<<endl;

	scanner::ptr_type s_ptr(new scanner(*pool, sm, lc, si));

	s_ptr->async_get_next_row(boost::bind(on_get_next_row, s_ptr, _1, _2));

	// 结束
	std::string input;
	while(std::cin >> input)
	{
		if(input == "close")
		{
			sm.close();
		}
		else if(input == "quit")
		{
			break;
		}
	}

	pool->stop();
	pool->join();

	return 0;
}
